/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * openGauss is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 * http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * ---------------------------------------------------------------------------------------
 * imcu_cache_mgr.cpp
 * routines to support ColStore
 *
 * IDENTIFICATION
 * src/gausskernel/storage/imcstore/imcu_cache_mgr.cpp
 * ---------------------------------------------------------------------------------------
 */
#include "postgres.h"
#include "knl/knl_variable.h"
#include "access/imcs/imcu_cache_mgr.h"
#include "utils/aiomem.h"
#include "executor/instrument.h"
#include "utils/resowner.h"
#include "storage/ipc.h"
#include "miscadmin.h"
#include "pgxc/pgxc.h"
#include "nodes/pg_list.h"
#include "utils/dynahash.h"
#include "access/imcs/imcu_storage.h"
#include "catalog/gs_imcs.h"
#include "access/imcs/imcs_ctlg.h"

#include "storage/buf/block.h"
#include "executor/executor.h"
#include "access/tupdesc.h"
#include "access/tableam.h"

/* FUTURE CASE: imcstore_buffers and IMCUCacheMode should be tunable */
int im_cu_cache_mode = 1;

#define IM_BUILD_BUG_ON_CONDITION(condition) ((void)sizeof(char[1 - 2 * (condition)]))

IMCUCacheMgr *IMCUCacheMgr::imcu_data_cache = NULL;

/*
 * @Description: IMCStoreDataCacheMgrNumLocks
 * Returns the number of LW locks required by the DataCacheMgr instance.
 * This function is called by NumLWLocks() to calculate the required memory
 * for all the LW Locks.  Since this must be done prior to initializing the
 * instance of the DataCacheMgr class, the function cannot be defined
 * as a method of the class.
 * @Return: lock number
 * @See also:
 */
int IMCStoreDataCacheMgrNumLocks()
{
    int64 cache_size = IMCacheMgrCalcSizeByType(MGR_CACHE_TYPE_DATA);
    return IMCacheMgrNumLocks(cache_size, BLCKSZ);
}

void MakeGlobalIMCUPath()
{
    const char *globaldir = "global/imcu";
    if (mkdir(globaldir, S_IRWXU) < 0) {
        if (errno == EEXIST) {
            return;
        }

        if (mkdir(globaldir, S_IRWXU) < 0 && errno != EEXIST) {
            ereport(ERROR, (errmodule(MOD_CACHE), errmsg("cannot make imcu directory \"%s\": %m", globaldir)));
        }
    }
}

void MakeBaseIMCUPath()
{
    const char *basedir = "base/imcu";
    if (mkdir(basedir, S_IRWXU) < 0) {
        if (errno == EEXIST) {
            return;
        }

        if (mkdir(basedir, S_IRWXU) < 0 && errno != EEXIST) {
            ereport(ERROR, (errmodule(MOD_CACHE), errmsg("cannot make imcu directory \"%s\": %m", basedir)));
        }
    }
}

void RemoveGlobalIMCUFile()
{
    DIR *dir;
    dir = opendir("global/imcu");
    if (dir == NULL) {
        MakeGlobalIMCUPath();
        return;
    }

    errno_t ss_rc;
    struct dirent *de = NULL;
    char path_buf[MAXPGPATH] = {0};
    struct stat st;
    while ((de = readdir(dir)) != NULL) {
        CHECK_FOR_INTERRUPTS();
        if ((strcmp(de->d_name, ".") == 0 || strcmp(de->d_name, "..") == 0)) {
            continue;
        }
        ss_rc = snprintf_s(path_buf, MAXPGPATH, MAXPGPATH - 1, "%s/%s", "global/imcu", de->d_name);
        securec_check_ss_c(ss_rc, "\0", "\0");
        if (lstat(path_buf, &st) < 0) {
            ereport(ERROR, (errmodule(MOD_CACHE), errmsg("cannot remove imcu file \"%s\": %m", path_buf)));
            continue;
        }
        /* only find file */
        if (S_ISREG(st.st_mode)) {
            remove(path_buf);
        } else {
            continue;
        }
    }

    (void)closedir(dir);
}

void RemoveBaseIMCUFile()
{
    DIR *dir;
    dir = opendir("base/imcu");
    if (dir == NULL) {
        MakeBaseIMCUPath();
        return;
    }

    errno_t ss_rc;
    struct dirent *de = NULL;
    char path_buf[MAXPGPATH] = {0};
    struct stat st;
    while ((de = readdir(dir)) != NULL) {
        CHECK_FOR_INTERRUPTS();
        if ((strcmp(de->d_name, ".") == 0 || strcmp(de->d_name, "..") == 0)) {
            continue;
        }
        ss_rc = snprintf_s(path_buf, MAXPGPATH, MAXPGPATH - 1, "%s/%s", "base/imcu", de->d_name);
        securec_check_ss_c(ss_rc, "\0", "\0");
        if (lstat(path_buf, &st) < 0) {
            ereport(ERROR, (errmodule(MOD_CACHE), errmsg("cannot remove imcu file \"%s\": %m", path_buf)));
            continue;
        }
        /* only find file */
        if (S_ISREG(st.st_mode)) {
            remove(path_buf);
        } else {
            continue;
        }
    }

    (void)closedir(dir);
}

void RemoveIMCUFile()
{
    RemoveGlobalIMCUFile();
    RemoveBaseIMCUFile();
}

/*
 * @Description: get data cache manage current memory cache used size
 * @Return: cache used size
 * @See also:
 */
int64 IMCUCacheMgr::GetCurrentMemSize()
{
    return m_cache_mgr->GetCurrentMemSize();
}

/*
 * @Description: create or recreate the Singleton Instance of IMCU cache
 * @See also:
 */
void IMCUCacheMgr::NewSingletonInstance(void)
{
    int64 cache_size = 0;
    /* only Postmaster has the privilege to create or recreate IMCU cache instance */
    if (IsUnderPostmaster)
        return;

    int condition = sizeof(DataSlotTagKey) != MAX_CACHE_TAG_LEN ? 1 : 0;
    IM_BUILD_BUG_ON_CONDITION(condition);

    if (imcu_data_cache == NULL) {
        /* create this instance at the first time */
        imcu_data_cache = New(CurrentMemoryContext)IMCUCacheMgr;
        imcu_data_cache->m_cache_mgr = New(CurrentMemoryContext)IMCSCacheMgr;
    } else {
        /* destroy all resources of its members */
        imcu_data_cache->m_cache_mgr->Destroy();
        SpinLockFree(&imcu_data_cache->m_adio_write_cache_lock);
    }
    cache_size = IMCacheMgrCalcSizeByType(MGR_CACHE_TYPE_DATA);
    imcu_data_cache->m_cstoreMaxSize = cache_size;
    SpinLockInit(&imcu_data_cache->m_adio_write_cache_lock);
    /* init or reset this instance */
    IMCSCacheMgr *imcs_cache_mgr = dynamic_cast<IMCSCacheMgr *>(imcu_data_cache->m_cache_mgr);
    imcs_cache_mgr->Init(cache_size, BLCKSZ, MGR_CACHE_TYPE_DATA, sizeof(IMCU));
    ereport(LOG, (errmodule(MOD_CACHE), errmsg("set data cache  size(%ld)", cache_size)));

    HASHCTL info;
    errno_t rc = memset_s(&info, sizeof(info), 0, sizeof(info));
    securec_check(rc, "\0", "\0");

    imcu_data_cache->m_imcs_context = AllocSetContextCreate(g_instance.instance_context, "ImcsContext",
        ALLOCSET_SMALL_MINSIZE, ALLOCSET_SMALL_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE, SHARED_CONTEXT);

    /* BufferTag maps to Buffer */
    info.keysize = sizeof(Oid);
    info.entrysize = sizeof(IMCSDesc_Entry);
    info.hash = tag_hash;
    info.hcxt = imcu_data_cache->m_imcs_context;
    info.num_partitions = NUM_CACHE_BUFFER_PARTITIONS / 2;

    imcu_data_cache->m_imcs_latch = LWLockAssign(LWTRANCHE_IMCS_DESC_LATCH);
    imcu_data_cache->m_imcs_hash = hash_create("IMCS_Desc Lookup Table", 40960, &info,
        HASH_ELEM | HASH_FUNCTION | HASH_PARTITION | HASH_CONTEXT | HASH_EXTERN_CONTEXT);
    imcu_data_cache->m_xlog_current_lsn = InvalidXLogRecPtr;
    imcu_data_cache->m_startup_cur_lsn = InvalidXLogRecPtr;
    imcu_data_cache->m_xlog_lsn_lock = LWLockAssign(LWTRANCHE_HTAP_XLOG_LSN_LOCK);
    imcu_data_cache->m_imcs_hash = hash_create("IMCS_Desc Lookup Table", 40960, &info,
        HASH_ELEM | HASH_FUNCTION | HASH_PARTITION | HASH_CONTEXT | HASH_EXTERN_CONTEXT);

    RemoveIMCUFile();
}

void IMCUCacheMgr::DropRelationImcuCache(const RelFileNode &rnode, int colId)
{
    CacheTag tag = { 0 };
    CUSlotTag cu_tag = { { InvalidOid, InvalidOid, InvalidOid }, 0, 0, 0, 0 };
    const int max_slot = m_cache_mgr->GetUsedCacheSlotNum();

    for (CacheSlotId_t slot = 0; slot <= max_slot; slot++) {
        m_cache_mgr->CopyCacheBlockTag(slot, &tag);
        cu_tag = *(CUSlotTag *)tag.key;
        if (CACHE_COlUMN_DATA == tag.type && RelFileColumnNodeRelEquals(rnode, cu_tag.m_rnode) &&
            colId == cu_tag.m_colId) {
            /* try to invalid this IMCU */
            InvalidateCU(&cu_tag.m_rnode, cu_tag.m_colId, cu_tag.m_CUId, cu_tag.m_cuPtr);
        }
    }
}

/*
 * @Description:  get cu cached buffer
 * @IN cuSlotId: slot id
 * @Return: IMCU data pointer
 * @See also:
 */
IMCU *IMCUCacheMgr::GetCUBuf(int cuSlotId)
{
    return (IMCU *)m_cache_mgr->GetCacheBlock(cuSlotId);
}


/*
 * @Description: add imcu data block into data cache manager
 * @IN dataSlotTag: data slot tag
 * @IN hasFound: whether found or not
 * @IN size: need block size
 * @Return: slot id
 * @See also:
 */
CacheSlotId_t IMCUCacheMgr::ReserveImcuBlock(DataSlotTag *dataSlotTag, IMCUDesc *cuDescPtr, bool &hasFound, IMCU *imcu)
{
    IMCU *cu = NULL;
    errno_t rc = EOK;
    CacheSlotId_t slot_id = CACHE_BLOCK_INVALID_IDX;
    slot_id = ReserveDataBlock(dataSlotTag, cuDescPtr->cu_size, hasFound);

    if (hasFound) {
        UnPinDataBlock(slot_id);
        return slot_id;
    }

    cu = (IMCU *)m_cache_mgr->GetCacheBlock(slot_id);

    cu->m_srcBuf = imcu->m_srcBuf;
    cu->m_nulls = imcu->m_nulls;
    cu->m_srcData = imcu->m_srcData;

    char *buf = (char *)malloc(imcu->m_compressedBufSize);
    rc = memcpy_s(buf, imcu->m_compressedBufSize, imcu->m_compressedBuf, imcu->m_compressedBufSize);
    securec_check(rc, "\0", "\0");
    cu->m_compressedBuf = buf;

    cu->m_offset = imcu->m_offset;
    cu->m_tmpinfo = imcu->m_tmpinfo;
    cu->m_compressedLoadBuf = imcu->m_compressedLoadBuf;
    cu->m_head_padding_size = imcu->m_head_padding_size;
    cu->m_offsetSize = imcu->m_offsetSize;
    cu->m_srcBufSize = imcu->m_srcBufSize;
    cu->m_srcDataSize = imcu->m_srcDataSize;
    cu->m_compressedBufSize = imcu->m_compressedBufSize;
    cu->m_cuSize = imcu->m_cuSize;
    cu->m_cuSizeExcludePadding = imcu->m_cuSizeExcludePadding;
    cu->m_crc = imcu->m_crc;
    cu->m_magic = imcu->m_magic;
    cu->m_eachValSize = imcu->m_eachValSize;
    cu->m_typeMode = imcu->m_typeMode;
    cu->m_bpNullRawSize = imcu->m_bpNullRawSize;
    cu->m_bpNullCompressedSize = imcu->m_bpNullCompressedSize;
    cu->m_infoMode = imcu->m_infoMode;
    cu->m_atttypid = imcu->m_atttypid;
    cu->m_adio_error = imcu->m_adio_error;
    cu->m_cache_compressed = imcu->m_cache_compressed;
    cu->m_inCUCache = true;
    cu->m_numericIntLike = imcu->m_numericIntLike;
    cuDescPtr->m_cu = cu;
    FREE_POINTER(imcu);
    UnPinDataBlock(slot_id);
    DataBlockCompleteIO(slot_id);

    return slot_id;
}

int IMCUCacheMgr::GetCacheSlotNum()
{
    return this->m_cache_mgr->GetUsedCacheSlotNum();
}

CacheDesc *IMCUCacheMgr::GetCacheSlotDesc(int cuSlotId)
{
    IMCSCacheMgr *imcs_cache_mgr = dynamic_cast<IMCSCacheMgr *>(this->m_cache_mgr);
    return imcs_cache_mgr->GetCacheDesc(cuSlotId);
}

/*
 * @Description: reserve  data block from data cache manager
 * @IN dataSlotTag: data slot tag
 * @IN hasFound: whether found or not
 * @IN size: need block size
 * @Return: slot id
 * @See also:
 */
CacheSlotId_t IMCUCacheMgr::ReserveDataBlock(DataSlotTag *dataSlotTag, int size, bool &hasFound)
{
    CacheSlotId_t slot = CACHE_BLOCK_INVALID_IDX;
    CacheTag cache_tag = { 0 };

    IMCSCacheMgr *imcs_cache_mgr = dynamic_cast<IMCSCacheMgr *>(m_cache_mgr);
    imcs_cache_mgr->InitCacheBlockTag(&cache_tag, dataSlotTag->slotType, &dataSlotTag->slotTag, sizeof(DataSlotTagKey));
    slot = imcs_cache_mgr->ReserveCacheBlock(&cache_tag, size, hasFound);
    if (!hasFound) {
        /* remember block slot in process */
        Assert(!IsValidCacheSlotID(t_thrd.storage_cxt.ImcuCacheBlockInProgressIO));
        t_thrd.storage_cxt.ImcuCacheBlockInProgressIO = slot;
    }
    if (cache_tag.type == CACHE_COlUMN_DATA) {
        CUSlotTag *cuslot_tag = &dataSlotTag->slotTag.cuSlotTag;
        ereport(DEBUG1, (errmodule(MOD_CACHE),
            errmsg("allocate metadata block, type(%d), slot_id(%d), spcID(%u), dbID(%u), relID(%u), col_id(%d), "
            "cuID(%d), cuPoint(%lu)",
            cache_tag.type, slot, cuslot_tag->m_rnode.spcNode, cuslot_tag->m_rnode.dbNode, cuslot_tag->m_rnode.relNode,
            cuslot_tag->m_colId, cuslot_tag->m_CUId, cuslot_tag->m_cuPtr)));
    } else if (cache_tag.type == CACHE_ORC_DATA) {
        ORCSlotTag *orcslot_tag = &dataSlotTag->slotTag.orcSlotTag;
        ereport(DEBUG1, (errmodule(MOD_CACHE),
            errmsg("allocate metadata block, type(%d), slot_id(%d), spcID(%u), dbID(%u), relID(%u), fileID(%d), "
            "length(%lu), offset(%lu)",
            cache_tag.type, slot, orcslot_tag->m_rnode.spcNode, orcslot_tag->m_rnode.dbNode,
            orcslot_tag->m_rnode.relNode, orcslot_tag->m_fileId, orcslot_tag->m_length, orcslot_tag->m_offset)));
    } else if (cache_tag.type == CACHE_OBS_DATA) {
        OBSSlotTag *obsslot_tag = &dataSlotTag->slotTag.obsSlotTag;
        ereport(DEBUG1, (errmodule(MOD_CACHE),
            errmsg("allocate metadata block, type(%d), slot_id(%d), serverHash(%u), bucketHash(%u), fileHash1( %u), "
            "fileHash2(%u), length(%lu), offset(%lu)",
            cache_tag.type, slot, obsslot_tag->m_serverHash, obsslot_tag->m_bucketHash, obsslot_tag->m_fileFirstHash,
            obsslot_tag->m_fileSecondHash, obsslot_tag->m_length, obsslot_tag->m_offset)));
    }

    return slot;
}


void IMCUCacheMgr::AcquireImcsDescLock(_in_ Oid imcs_oid, LWLockMode lock_mode)
{
    IMCSDesc *result = GetImcsDesc(imcs_oid);
    Assert(result != NULL);
    LWLockAcquire(result->imcs_desc_lock, lock_mode);
}

void IMCUCacheMgr::ReleaseImcsDescLock(_in_ Oid imcs_oid)
{
    IMCSDesc *result = GetImcsDesc(imcs_oid);
    Assert(result != NULL);
    LWLockRelease(result->imcs_desc_lock);
}

void IMCUCacheMgr::CreateImcsRgLock(_in_ Oid imcs_oid, int rg)
{
    IMCSDesc *result = GetImcsDesc(imcs_oid);
    Assert(result != NULL);

    MemoryContext oldcontext = MemoryContextSwitchTo(IMCU_CACHE->m_imcs_context);
    pthread_rwlock_t *new_rg_lock = (pthread_rwlock_t *)palloc0(sizeof(pthread_rwlock_t));
    (void)MemoryContextSwitchTo(oldcontext);

    if (!new_rg_lock) {
        ereport(ERROR, (errmsg("Failed to create lock")));
    }
    PthreadRwLockInit(new_rg_lock, NULL);

    SpinLockAcquire(&result->imcs_rg_lock);
    result->rg_lock_map[rg] = new_rg_lock;
    SpinLockRelease(&result->imcs_rg_lock);
}


bool IMCUCacheMgr::AcquireImcsRgLock(_in_ Oid imcs_oid, LWLockMode lock_mode, int rg)
{
    pthread_rwlock_t *lock;
    IMCSDesc *result = GetImcsDesc(imcs_oid);
    Assert(result != NULL);

    SpinLockAcquire(&result->imcs_rg_lock);
    auto it = result->rg_lock_map.find(rg);
    SpinLockRelease(&result->imcs_rg_lock);

    if (it == result->rg_lock_map.end()) {
        return false;
    }
    lock = it->second;
    if (lock_mode == LW_EXCLUSIVE)
        PthreadRWlockWrlock(t_thrd.utils_cxt.CurrentResourceOwner, lock);
    else
        PthreadRWlockRdlock(t_thrd.utils_cxt.CurrentResourceOwner, lock);

    return true;
}

void IMCUCacheMgr::ReleaseImcsRgLock(_in_ Oid imcs_oid, int rg)
{
    IMCSDesc *result = GetImcsDesc(imcs_oid);
    Assert(result != NULL);

    SpinLockAcquire(&result->imcs_rg_lock);
    auto it = result->rg_lock_map.find(rg);
    SpinLockRelease(&result->imcs_rg_lock);

    if (it == result->rg_lock_map.end()) {
        ereport(ERROR, (errmsg("Failed to find lock for Oid %d, rg %d", imcs_oid, rg)));
    }
    pthread_rwlock_t *lock = it->second;
    PthreadRWlockUnlock(t_thrd.utils_cxt.CurrentResourceOwner, lock);
}


/*
 * @Description: DataBlockCompleteIO
 * When an I/O is finished mark the CU as not busy and wake the next waiter.
 * this method is simmilar to the column store buffer_terminate_io
 * @Param[IN] slot_id: slot id
 * @See also:
 */
void IMCUCacheMgr::DataBlockCompleteIO(CacheSlotId_t slot_id)
{
    m_cache_mgr->CompleteIO(slot_id);

    /* clear block slot in IO progress, aio completer thread does not use it */
    t_thrd.storage_cxt.ImcuCacheBlockInProgressIO = CACHE_BLOCK_INVALID_IDX;

    return;
}

/*
 * @Description: IMCU cache will uncompress IMCU raw data.
 * @Param[IN] cuDescPtr: IMCU desc info
 * @Param[IN] slotId: IMCU slot id
 * @Return: CUUncompressedRetCode value
 * @See also:
 */
CUUncompressedRetCode IMCUCacheMgr::StartUncompressCU(IMCUDesc *cuDescPtr, CacheSlotId_t slotId, int planNodeId,
    bool timing, int align_size)
{
    IMCU *cu_ptr = GetCUBuf(slotId);

    m_cache_mgr->AcquireCompressLock(slotId);

    /*
     * if the slot flag is set to CACHE_BLOCK_ERROR by some abort threads,
     * the compressed buffer is also valid no matter the IMCU comes from primary
     * node or remote mode. So, the thread can continue to uncompress.
     */
    if (m_cache_mgr->IsIOBusy(slotId)) {
        /*
         * the IMCU is being reloading by remote read thread,
         * return CU_RELOADING and retry to load IMCU.
         */
        m_cache_mgr->ReleaseCompressLock(slotId);
        return CU_RELOADING;
    }

    if (!cu_ptr->m_cache_compressed) {
        /* another therad have decompressed this IMCU data */
        m_cache_mgr->ReleaseCompressLock(slotId);
        return CU_OK;
    }

    if (cu_ptr->m_adio_error) {
        /* IO error */
        this->m_cache_mgr->ReleaseCompressLock(slotId);
        return CU_ERR_ADIO;
    }

    /* remember this slot id */
    Assert(!IsValidCacheSlotID(t_thrd.storage_cxt.CacheBlockInProgressUncompress));
    t_thrd.storage_cxt.CacheBlockInProgressUncompress = slotId;

    if (cu_ptr->CheckCrc() == false) {
        /* CRC check failed */
        m_cache_mgr->ReleaseCompressLock(slotId);
        return CU_ERR_CRC;
    }

    if (cu_ptr->CheckMagic(cuDescPtr->magic) == false) {
        m_cache_mgr->ReleaseCompressLock(slotId);
        return CU_ERR_MAGIC;
    }

    /* macro for tracing cstore scan */
#define UNCOMPRESS_TRACE(A)     \
    do {                        \
        if (unlikely(timing)) { \
            A;                  \
        }                       \
    } while (0)

    /* Always presume compressed disk and uncompressed cache. */
    UNCOMPRESS_TRACE(TRACK_START(planNodeId, UNCOMPRESS_CU));
    cu_ptr->UnCompress(cuDescPtr->row_count, cuDescPtr->magic, align_size);
    UNCOMPRESS_TRACE(TRACK_END(planNodeId, UNCOMPRESS_CU));

    /* Do not put the compressedBuf in the cache
     * In the future we may have compressed data in the cache or let the
     * caller choose, but for now we just keep the uncompressed data.
     */
    cu_ptr->FreeCompressBuf();

    /* Adjust the allocation reservation to take into account
     * compression or expansion.
     */
    int cu_uncompress_size = cu_ptr->GetUncompressBufSize();
    m_cache_mgr->AdjustCacheMem(slotId, cuDescPtr->cu_size, cu_uncompress_size);
    m_cache_mgr->ReleaseCompressLock(slotId);

    TerminateCU(false);
    return CU_OK;
}

/*
 * @Description: SaveImcsDesc
 * @See also:
 */
void IMCUCacheMgr::SaveImcsDesc(_in_ Oid imcs_oid)
{
    LWLockAcquire(m_imcs_latch, LW_EXCLUSIVE);
    IMCSDesc_Entry *result;
    errno_t rc = EOK;
    bool has_found = false;
    MemoryContext oldcontext = NULL;
    Assert(m_imcs_hash != NULL);
    oldcontext = MemoryContextSwitchTo(m_imcs_hash->hcxt);

    result = (IMCSDesc_Entry *)hash_search(m_imcs_hash, &imcs_oid, HASH_ENTER, &has_found);
    if (!has_found) {
        HASHCTL info;
        rc = memset_s(&info, sizeof(info), 0, sizeof(info));
        securec_check(rc, "\0", "\0");

        /* BufferTag maps to Buffer */
        char context_name[150] = {0};
        rc = snprintf_s(context_name, sizeof(context_name), 149, "ImcuDescHashtblContext%d", imcs_oid);
        securec_check_ss(rc, "\0", "\0");

        MemoryContext context = AllocSetContextCreate(m_imcs_hash->hcxt, context_name, ALLOCSET_SMALL_MINSIZE,
            ALLOCSET_SMALL_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE, SHARED_CONTEXT);

        info.keysize = sizeof(IMCUDesc_Tag);
        info.entrysize = sizeof(IMCUDesc_Entry);
        info.hash = tag_hash;
        info.hcxt = context;
        info.num_partitions = NUM_CACHE_BUFFER_PARTITIONS / 2;

        char *buf = (char *)palloc(sizeof(IMCSDesc));
        if (!buf) {
            ereport(FATAL, (errmsg("IMCSDesc allocation of size %ld Failed", sizeof(IMCSDesc))));
        }
        result->imcs_desc = new (buf)IMCSDesc();
        result->imcs_desc->imcs_oid = imcs_oid;
        result->imcs_desc->max_cu_id = 0;
        result->imcs_desc->rowgroup_num = 0;
#ifdef ENABLE_HTAP
        result->imcs_desc->deletedList = NULL;
#endif
        result->imcs_desc->last_flushed_csn = 0;
        result->imcs_desc->cu_num_in_mem = 0;
        result->imcs_desc->cu_mem_size = 0;
        result->imcs_desc->cu_num_in_disk = 0;
        result->imcs_desc->cu_disk_size = 0;
        result->imcs_desc->imcs_desc_lock = LWLockAssign(LWTRANCHE_IMCS_DESC_LOCK);
        result->imcs_desc->imcu_desc_latch = LWLockAssign(LWTRANCHE_IMCU_DESC_LATCH);
        SpinLockInit(&result->imcs_desc->imcs_rg_lock);
        result->imcs_desc->im_cudesc_table = hash_create("IMCU_Desc col Lookup Table", 40960, &info,
            HASH_ELEM | HASH_FUNCTION | HASH_PARTITION | HASH_CONTEXT | HASH_EXTERN_CONTEXT);
    }

    LWLockRelease(m_imcs_latch);
    (void)MemoryContextSwitchTo(oldcontext);
}

uint32 IMCUCacheMgr::SaveImcuDesc(_in_ Oid imcs_oid, _in_ IMCUDesc *cuDescPtr, _in_ int attnum, _in_ int attlen)
{
    bool has_found = false;
    errno_t rc = EOK;
    IMCSDesc_Entry *result;
    MemoryContext oldcontext = NULL;
    IMCUDesc_Tag cu_desc_tag;
    cu_desc_tag.attnum = attnum;
    cu_desc_tag.cu_id = cuDescPtr->cu_id;
    Assert(m_imcs_hash != NULL);
    oldcontext = MemoryContextSwitchTo(m_imcs_hash->hcxt);
    uint32 hash_code = 0;

    LWLockAcquire(m_imcs_latch, LW_SHARED);
    hash_code = get_hash_value(m_imcs_hash, &imcs_oid);
    result = (IMCSDesc_Entry *)hash_search(m_imcs_hash, &imcs_oid, HASH_FIND, &has_found);
    LWLockRelease(m_imcs_latch);
    Assert(result != NULL);

    uint32 max_cu_id = result->imcs_desc->max_cu_id;
    LWLockAcquire(result->imcs_desc->imcu_desc_latch, LW_EXCLUSIVE);
    result->imcs_desc->cu_num_in_mem++;
    result->imcs_desc->cu_mem_size += cuDescPtr->cu_size;
    if (cuDescPtr->cu_id > max_cu_id) {
        max_cu_id = cuDescPtr->cu_id;
    }

    (void)MemoryContextSwitchTo(result->imcs_desc->im_cudesc_table->hcxt);

    IMCUDesc_Entry *m_cudesc_info =
        (IMCUDesc_Entry *)hash_search(result->imcs_desc->im_cudesc_table, &cu_desc_tag, HASH_ENTER, &has_found);

    LWLockRelease(result->imcs_desc->imcu_desc_latch);
    bool need_length_info = attlen < 0;
    char *val_ptr = NULL;
    IMCUDesc *deep_copy = (IMCUDesc *)palloc(sizeof(IMCUDesc));

    /* Put min value into cudesc->min */
    char *min_ptr = deep_copy->cu_min;
    int len_1 = 48;
    val_ptr = cuDescPtr->cu_min;
    if (need_length_info) {
        *min_ptr = val_ptr[0];
        min_ptr = min_ptr + 1;
        len_1 -= 1;
        rc = memcpy_s(min_ptr, len_1, val_ptr + 1, val_ptr[0]);
        securec_check(rc, "", "");
    } else {
        rc = memcpy_s(min_ptr, len_1, val_ptr, attlen);
        securec_check(rc, "", "");
    }

    /* Put max value into cudesc->max */
    char *max_ptr = deep_copy->cu_max;
    int len_2 = 48;
    val_ptr = cuDescPtr->cu_max;
    if (need_length_info) {
        *max_ptr = val_ptr[0];
        max_ptr = max_ptr + 1;
        len_2 -= 1;
        rc = memcpy_s(max_ptr, len_2, val_ptr + 1, val_ptr[0]);
        securec_check(rc, "", "");
    } else {
        rc = memcpy_s(max_ptr, len_2, val_ptr, attlen);
        securec_check(rc, "", "");
    }

    deep_copy->cu_id = cuDescPtr->cu_id;
    deep_copy->cu_mode = cuDescPtr->cu_mode;
    deep_copy->cu_pointer = cuDescPtr->cu_pointer;
    deep_copy->cu_size = cuDescPtr->cu_size;
    deep_copy->xmin = cuDescPtr->xmin;
    deep_copy->row_count = cuDescPtr->row_count;
    deep_copy->magic = cuDescPtr->magic;
    deep_copy->m_cu = cuDescPtr->m_cu;
    deep_copy->flag = cuDescPtr->flag;

    m_cudesc_info->cu_tag.cu_id = cuDescPtr->cu_id;
    m_cudesc_info->cu_tag.attnum = attnum;
    m_cudesc_info->cu_desc = deep_copy;
    (void)MemoryContextSwitchTo(oldcontext);
    return max_cu_id;
}


IMCUDesc *IMCUCacheMgr::GetImcuDesc(_in_ Oid imcs_oid, _in_ uint32 cu_id, _in_ int attnum)
{
    IMCUDesc_Tag cu_desc_tag;
    cu_desc_tag.attnum = attnum;
    cu_desc_tag.cu_id = cu_id;
    Assert(m_imcs_hash != NULL);
    IMCSDesc_Entry *result;

    LWLockAcquire(m_imcs_latch, LW_SHARED);
    result = (IMCSDesc_Entry *)hash_search(m_imcs_hash, &imcs_oid, HASH_FIND, NULL);
    LWLockRelease(m_imcs_latch);
    if (result == NULL) {
        return NULL;
    }

    IMCUDesc_Entry *cudesc_info;
    LWLockAcquire(result->imcs_desc->imcu_desc_latch, LW_SHARED);
    cudesc_info = (IMCUDesc_Entry *)hash_search(result->imcs_desc->im_cudesc_table, &cu_desc_tag, HASH_FIND, NULL);
    LWLockRelease(result->imcs_desc->imcu_desc_latch);
    if (cudesc_info == NULL) {
        return NULL;
    }
    return cudesc_info->cu_desc;
}

void IMCUCacheMgr::IncreaseIMCUDesc(_in_ Oid imcs_oid, _in_ uint32 cu_id, _in_ int attnum)
{
    IMCUDesc_Tag cu_desc_tag;
    cu_desc_tag.attnum = attnum;
    cu_desc_tag.cu_id = cu_id;
    Assert(m_imcs_hash != NULL);
    IMCSDesc_Entry *result;

    LWLockAcquire(m_imcs_latch, LW_SHARED);
    result = (IMCSDesc_Entry *)hash_search(m_imcs_hash, &imcs_oid, HASH_FIND, NULL);
    LWLockRelease(m_imcs_latch);
    if (result == NULL) {
        return;
    }

    LWLockAcquire(result->imcs_desc->imcu_desc_latch, LW_EXCLUSIVE);
    result->imcs_desc->cu_num_in_mem++;
    result->imcs_desc->cu_num_in_disk--;
    if (result->imcs_desc->cu_num_in_disk == 0) {
        UpdateImcHashTblStatus(imcs_oid, IMCS_STATUS_COMPLETED);
    }
    LWLockRelease(result->imcs_desc->imcu_desc_latch);

    IMCUDesc_Entry *cudesc_info;
    LWLockAcquire(result->imcs_desc->imcu_desc_latch, LW_EXCLUSIVE);
    cudesc_info = (IMCUDesc_Entry *)hash_search(result->imcs_desc->im_cudesc_table, &cu_desc_tag, HASH_FIND, NULL);
    if (cudesc_info != NULL) {
        cudesc_info->cu_desc->flag = IMCUDescInMemory;
    }
    LWLockRelease(result->imcs_desc->imcu_desc_latch);
}

void IMCUCacheMgr::IncreaseImcsRowGroupNum(_in_ Oid imcs_oid, _in_ uint32 max_cu_id)
{
    bool has_found = false;
    IMCSDesc_Entry *result;

    LWLockAcquire(m_imcs_latch, LW_SHARED);
    result = (IMCSDesc_Entry *)hash_search(m_imcs_hash, &imcs_oid, HASH_FIND, &has_found);
    LWLockRelease(m_imcs_latch);
    Assert(result != NULL);
#ifdef ENABLE_HTAP
    result->imcs_desc->increaseRGNum();
    if (max_cu_id > result->imcs_desc->max_cu_id) {
        result->imcs_desc->max_cu_id = max_cu_id;
    }
#else
    LWLockAcquire(result->imcs_desc->imcu_desc_latch, LW_EXCLUSIVE);
    result->imcs_desc->rowgroup_num++;
    if (max_cu_id > result->imcs_desc->max_cu_id) {
        result->imcs_desc->max_cu_id = max_cu_id;
    }
    LWLockRelease(result->imcs_desc->imcu_desc_latch);
#endif
}

IMCUDesc *IMCUCacheMgr::GetImcuDescByHash(_in_ IMCSDesc *imcs_desc, _in_ uint32 cu_id, _in_ int attnum)
{
    IMCUDesc_Tag cu_desc_tag;
    cu_desc_tag.attnum = attnum;
    cu_desc_tag.cu_id = cu_id;
    IMCUDesc_Entry *cudesc_info;

    LWLockAcquire(imcs_desc->imcu_desc_latch, LW_SHARED);
    cudesc_info = (IMCUDesc_Entry *)hash_search(imcs_desc->im_cudesc_table, &cu_desc_tag, HASH_FIND, NULL);
    LWLockRelease(imcs_desc->imcu_desc_latch);
    if (cudesc_info == NULL) {
        return NULL;
    }
    return cudesc_info->cu_desc;
}

void IMCUCacheMgr::DropImcuDesc(_in_ Oid imcs_oid, _in_ uint32 cu_id, _in_ int attnum)
{
    IMCUDesc_Tag cu_desc_tag;
    cu_desc_tag.attnum = attnum;
    cu_desc_tag.cu_id = cu_id;

    Assert(m_imcs_hash != NULL);
    IMCSDesc_Entry *result;
    LWLockAcquire(m_imcs_latch, LW_SHARED);
    result = (IMCSDesc_Entry *)hash_search(m_imcs_hash, &imcs_oid, HASH_FIND, NULL);
    LWLockRelease(m_imcs_latch);
    if (result == NULL) {
        return;
    }

    LWLockAcquire(result->imcs_desc->imcu_desc_latch, LW_EXCLUSIVE);
    IMCUDesc_Entry *cudesc_info =
        (IMCUDesc_Entry *)hash_search(result->imcs_desc->im_cudesc_table, &cu_desc_tag, HASH_REMOVE, NULL);
    IMCU *cu = (IMCU *)cudesc_info->cu_desc->m_cu;
    if (cu->m_inCUCache && !cu->m_cache_compressed) {
        result->imcs_desc->cu_num_in_mem--;
        result->imcs_desc->cu_mem_size -= cu->m_srcDataSize;
    }
    if (cu->m_inCUCache && cu->m_cache_compressed) {
        result->imcs_desc->cu_num_in_mem--;
        result->imcs_desc->cu_mem_size -= cu->m_cuSize;
    }
    pfree(cudesc_info);
    LWLockRelease(result->imcs_desc->imcu_desc_latch);
}

void IMCUCacheMgr::DropImcuColDesc(_in_ Oid imcs_oid, _in_ int attnum)
{
    MemoryContext oldcontext = NULL;
    Assert(m_imcs_hash != NULL);
    IMCSDesc_Entry *result;
    LWLockAcquire(m_imcs_latch, LW_SHARED);
    result = (IMCSDesc_Entry *)hash_search(m_imcs_hash, &imcs_oid, HASH_FIND, NULL);
    LWLockRelease(m_imcs_latch);
    if (result == NULL) {
        return;
    }
    oldcontext = MemoryContextSwitchTo(result->imcs_desc->im_cudesc_table->hcxt);
    LWLockAcquire(result->imcs_desc->imcu_desc_latch, LW_EXCLUSIVE);
    for (uint32 i = 0; i <= result->imcs_desc->max_cu_id; i++) {
        IMCUDesc_Tag cu_desc_tag;
        cu_desc_tag.attnum = attnum;
        cu_desc_tag.cu_id = i;
        IMCUDesc_Entry *cudesc_info =
            (IMCUDesc_Entry *)hash_search(result->imcs_desc->im_cudesc_table, &cu_desc_tag, HASH_REMOVE, NULL);
        IMCU *cu = (IMCU *)cudesc_info->cu_desc->m_cu;
        if (cu->m_inCUCache && !cu->m_cache_compressed) {
            result->imcs_desc->cu_num_in_mem--;
            result->imcs_desc->cu_mem_size -= cu->m_srcDataSize;
        }
        if (cu->m_inCUCache && cu->m_cache_compressed) {
            result->imcs_desc->cu_num_in_mem--;
            result->imcs_desc->cu_mem_size -= cu->m_cuSize;
        }
        pfree(cudesc_info);
    }
    LWLockRelease(result->imcs_desc->imcu_desc_latch);
    (void)MemoryContextSwitchTo(oldcontext);
}

IMCSDesc *IMCUCacheMgr::GetImcsDesc(_in_ Oid imcs_oid)
{
    LWLockAcquire(m_imcs_latch, LW_SHARED);
    Assert(m_imcs_hash != NULL);
    IMCSDesc_Entry *result;
    result = (IMCSDesc_Entry *)hash_search(m_imcs_hash, &imcs_oid, HASH_FIND, NULL);
    LWLockRelease(m_imcs_latch);
    if (result == NULL) {
        return NULL;
    }
    return result->imcs_desc;
}

void IMCUCacheMgr::DropImcsDesc(_in_ Oid imcs_oid)
{
    LWLockAcquire(m_imcs_latch, LW_EXCLUSIVE);
    Assert(m_imcs_hash != NULL);
    IMCSDesc_Entry *result;
    result = (IMCSDesc_Entry *)hash_search(m_imcs_hash, &imcs_oid, HASH_REMOVE, NULL);
    pg_atomic_fetch_sub_u32(&m_imcs_cnt, 1);
    LWLockRelease(m_imcs_latch);
    if (result == NULL) {
        return;
    }

    SpinLockFree(&result->imcs_desc->imcs_rg_lock);
    result->imcs_desc->~IMCSDesc();
    pfree((char *)result->imcs_desc);
}

void IMCUCacheMgr::DropRelationImcuFile(_in_ const RelFileNode &rnode, _in_ IMCSDesc *imcs_desc)
{
    IMCUDesc_Entry *imcu_entry;
    RelFileNodeOld m_rnode;
    m_rnode.dbNode = rnode.dbNode;
    m_rnode.spcNode = rnode.spcNode;
    m_rnode.relNode = rnode.relNode;

    CFileNode cFileNode(m_rnode);
    IMCUStorage *cu_storage = New(CurrentMemoryContext)IMCUStorage(cFileNode);
    LWLockAcquire(imcs_desc->imcu_desc_latch, LW_SHARED);

    HASH_SEQ_STATUS *imcu_hash_seq = (HASH_SEQ_STATUS *)palloc0(sizeof(HASH_SEQ_STATUS));
    hash_seq_init(imcu_hash_seq, imcs_desc->im_cudesc_table);
    imcu_entry = (IMCUDesc_Entry *)hash_seq_search(imcu_hash_seq);
    while (imcu_entry != NULL) {
        if (IsInDiskIMCU(imcu_entry->cu_desc->flag)) {
            cu_storage->RemoveImcuFile(imcu_entry->cu_tag.attnum, imcu_entry->cu_tag.cu_id);
        }
        imcu_entry = (IMCUDesc_Entry *)hash_seq_search(imcu_hash_seq);
    }
    cu_storage->Destroy();
    LWLockRelease(imcs_desc->imcu_desc_latch);
}

void IMCUCacheMgr::DropRelationColImcuFile(_in_ const RelFileNode &rnode, _in_ int col, _in_ IMCSDesc *imcs_desc)
{
    IMCUDesc_Entry *imcu_entry;
    RelFileNodeOld m_rnode;
    m_rnode.dbNode = rnode.dbNode;
    m_rnode.spcNode = rnode.spcNode;
    m_rnode.relNode = rnode.relNode;

    CFileNode cFileNode(m_rnode);
    IMCUStorage *cu_storage = New(CurrentMemoryContext)IMCUStorage(cFileNode);
    LWLockAcquire(imcs_desc->imcu_desc_latch, LW_EXCLUSIVE);

    HASH_SEQ_STATUS *imcu_hash_seq = (HASH_SEQ_STATUS *)palloc0(sizeof(HASH_SEQ_STATUS));
    hash_seq_init(imcu_hash_seq, imcs_desc->im_cudesc_table);
    imcu_entry = (IMCUDesc_Entry *)hash_seq_search(imcu_hash_seq);
    while (imcu_entry != NULL) {
        if (IsInDiskIMCU(imcu_entry->cu_desc->flag) && imcu_entry->cu_tag.attnum == col) {
            cu_storage->RemoveImcuFile(imcu_entry->cu_tag.attnum, imcu_entry->cu_tag.cu_id);
            imcs_desc->cu_num_in_disk--;
            imcs_desc->cu_disk_size -= imcu_entry->cu_desc->cu_size;
        }
        imcu_entry = (IMCUDesc_Entry *)hash_seq_search(imcu_hash_seq);
    }
    cu_storage->Destroy();
    LWLockRelease(imcs_desc->imcu_desc_latch);
}

/*
 * @Description: get Singleton Instance of IMCU cache
 * @Return: IMCU cache instance
 * @See also:
 */
IMCUCacheMgr *IMCUCacheMgr::GetInstance(void)
{
    Assert(imcu_data_cache != NULL);
    return imcu_data_cache;
}